package cn.boat.router.consumer;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.util.List;

public class RegisterCenterImpl implements RegisterCenterInterface {
    private String consumerGroupId; //消费者组标识
    private String subscribeTopic; //订阅的主题
    private List<String> subscribeTopics; // 订阅的多个主题，以列表的形式
    private String consumerID; // 消费者组内消费者标识
    private static final String CONSUMER_ROOT_PATH = "/boatMQ/consumers";
    private static final String IDS = "ids";
    // private static final S   tring OFFSETS = "offsets";
    private static final String OWNERS = "owners";
    private ZooKeeper zooKeeperConnection;
    private static boolean registerFlag = false;// 防止被多次注册
    /*public RegisterCenterImpl() {
        this.consumerGroupId ="";
        this.subscribeTopic = "";
        this.subscribeTopics = null;
        this.zooKeeperConnection =null;
        this.consumerID = null;
    }*/

    public RegisterCenterImpl(ZooKeeper zooKeeperConnection) {
        this.consumerGroupId ="";
        this.subscribeTopic = "";
        this.subscribeTopics = null;
        this.consumerID =null;
        this.zooKeeperConnection = zooKeeperConnection;
    }

    public void notExistsCreate(String consumerGroupId) throws Exception {
        Stat consumersExists = zooKeeperConnection.exists(CONSUMER_ROOT_PATH, false);
        if (consumersExists ==null){
            //第一次初始化的时候创建根路径 /consumers
            zooKeeperConnection.create(CONSUMER_ROOT_PATH,"consumers".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        Stat consumerGroupExists = zooKeeperConnection.exists(CONSUMER_ROOT_PATH + "/" + consumerGroupId, false);
        if (consumerGroupExists ==null){
            zooKeeperConnection.create(CONSUMER_ROOT_PATH + "/" + consumerGroupId,consumerGroupId.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        }
        Stat idsExists = zooKeeperConnection.exists(CONSUMER_ROOT_PATH + "/" + consumerGroupId + "/" + IDS, false);
        if (idsExists ==null){
            zooKeeperConnection.create(CONSUMER_ROOT_PATH + "/" + consumerGroupId + "/" + IDS,"ids".getBytes(),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        }
        Stat ownersExists = zooKeeperConnection.exists(CONSUMER_ROOT_PATH + "/" + consumerGroupId + "/" + OWNERS, false);
        if (ownersExists ==null){
            zooKeeperConnection.create(CONSUMER_ROOT_PATH + "/" + consumerGroupId + "/" + OWNERS,"owners".getBytes(),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        }

    }
// 如果消费者不给 consumer id zookeeper指定一个不重复的id
    @Override
    public void register(String consumerGroupId) throws Exception {
        // 加分布式锁
        MyLock myLock =new MyLock(zooKeeperConnection);
        myLock.acquireLock();
        notExistsCreate(consumerGroupId);
        myLock.releaseLock();
        if (registerFlag == true){
            System.out.println("您已经注册过请勿重复注册");
            return ;
        }
            String consumerIdPath = zooKeeperConnection.create(CONSUMER_ROOT_PATH + "/" + consumerGroupId + "/" + IDS + "/consumer",
                    "consumerid".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            // zooKeeperConnection.setData();
            zooKeeperConnection.setData(consumerIdPath,("consumer"+consumerIdPath.substring(consumerIdPath.length()-10)).getBytes(),-1);
            consumerID = ("consumer"+consumerIdPath.substring(consumerIdPath.length()-10));
            registerFlag =true;
        System.out.println(consumerID+"注册成功");
            // 等会看看是同步还是异步，要不要加锁

    }

    @Override
    public void register(String consumerGroupId, String consumerID) throws Exception {
        MyLock myLock =new MyLock(zooKeeperConnection);
        myLock.acquireLock();
        notExistsCreate(consumerGroupId);
        myLock.releaseLock();
        Stat exists = zooKeeperConnection.exists(CONSUMER_ROOT_PATH + "/" + consumerGroupId + IDS + "/" + consumerID, false);

        if (exists !=null){
            System.out.println("您已经注册过请勿重复注册");
            return;
        }
        String consumerIdPath = zooKeeperConnection.create(CONSUMER_ROOT_PATH + "/" + consumerGroupId + "/" + IDS + "/"+consumerID,
                    consumerID.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            // zooKeeperConnection.setData();
            /*zooKeeperConnection.setData(consumerIdPath,(consumerID.getBytes()),-1);*/
        this.consumerID =consumerID;
        registerFlag =true;
        System.out.println(consumerID+"注册成功");
        // 等会看看是同步还是异步，要不要加锁
    }

    public String getConsumerGroupId() {
        return consumerGroupId;
    }

    public String getSubscribeTopic() {
        return subscribeTopic;
    }

    public List<String> getSubscribeTopics() {
        return subscribeTopics;
    }

    public String getConsumerID() {
        return consumerID;
    }

    public ZooKeeper getZooKeeperConnection() {
        return zooKeeperConnection;
    }

    public static boolean isRegisterFlag() {
        return registerFlag;
    }
}
